diff --git a/sdk/storage/azdatalake/CHANGELOG.md b/sdk/storage/azdatalake/CHANGELOG.md
index bc7f4bc6c481..dee15c3860b5 100644
--- a/sdk/storage/azdatalake/CHANGELOG.md
+++ b/sdk/storage/azdatalake/CHANGELOG.md
@@ -5,6 +5,7 @@
 ### Features Added
 * Encryption Scope For SAS
 * CPK for Datalake
+* Create SubDirectory Client
 
 ### Breaking Changes
 
@@ -22,4 +23,4 @@
 
 ### Features Added
 
-* This is the initial preview release of the `azdatalake` library
+* This is the initial preview release of the `azdatalake` library
\ No newline at end of file
diff --git a/sdk/storage/azdatalake/assets.json b/sdk/storage/azdatalake/assets.json
index cb3cf3946254..15e28d300749 100644
--- a/sdk/storage/azdatalake/assets.json
+++ b/sdk/storage/azdatalake/assets.json
@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "go",
   "TagPrefix": "go/storage/azdatalake",
-  "Tag": "go/storage/azdatalake_c883370a69"
+  "Tag": "go/storage/azdatalake_ba5bfaa832"
 }
\ No newline at end of file
diff --git a/sdk/storage/azdatalake/directory/client.go b/sdk/storage/azdatalake/directory/client.go
index 97e907e11f7c..483a3608bff5 100644
--- a/sdk/storage/azdatalake/directory/client.go
+++ b/sdk/storage/azdatalake/directory/client.go
@@ -228,6 +228,29 @@ func (d *Client) NewFileClient(fileName string) (*file.Client, error) {
 	return (*file.Client)(base.NewPathClient(fileURL, newBlobURL, newBlobClient, d.generatedDirClientWithDFS().InternalClient().WithClientName(shared.FileClient), d.sharedKey(), d.identityCredential(), d.getClientOptions())), nil
 }
 
+// NewSubdirectoryClient creates a new directory.Client object by concatenating subdirectoryName to the end of this Client's URL.
+// The new directory.Client uses the same request policy pipeline as the Client.
+func (d *Client) NewSubdirectoryClient(subdirectoryName string) (*Client, error) {
+	subdirectoryName = url.PathEscape(strings.TrimRight(subdirectoryName, "/"))
+	subDirectoryURL := runtime.JoinPaths(d.DFSURL(), subdirectoryName)
+	newBlobURL, subDirectoryURL := shared.GetURLs(subDirectoryURL)
+	var newBlobClient *blockblob.Client
+	clientOptions := &blockblob.ClientOptions{ClientOptions: d.getClientOptions().ClientOptions}
+	var err error
+	if d.identityCredential() != nil {
+		newBlobClient, err = blockblob.NewClient(newBlobURL, *d.identityCredential(), clientOptions)
+	} else if d.sharedKey() != nil {
+		blobSharedKey, _ := exported.ConvertToBlobSharedKey(d.sharedKey())
+		newBlobClient, err = blockblob.NewClientWithSharedKeyCredential(newBlobURL, blobSharedKey, clientOptions)
+	} else {
+		newBlobClient, err = blockblob.NewClientWithNoCredential(newBlobURL, clientOptions)
+	}
+	if err != nil {
+		return nil, exported.ConvertToDFSError(err)
+	}
+	return (*Client)(base.NewPathClient(subDirectoryURL, newBlobURL, newBlobClient, d.generatedDirClientWithDFS().InternalClient().WithClientName(shared.DirectoryClient), d.sharedKey(), d.identityCredential(), d.getClientOptions())), nil
+}
+
 // Create creates a new directory.
 func (d *Client) Create(ctx context.Context, options *CreateOptions) (CreateResponse, error) {
 	lac, mac, httpHeaders, createOpts, cpkOpts := options.format()
diff --git a/sdk/storage/azdatalake/directory/client_test.go b/sdk/storage/azdatalake/directory/client_test.go
index 9a43fa362733..7214cdfd9726 100644
--- a/sdk/storage/azdatalake/directory/client_test.go
+++ b/sdk/storage/azdatalake/directory/client_test.go
@@ -72,7 +72,6 @@ type UnrecordedTestSuite struct {
 }
 
 func (s *UnrecordedTestSuite) TestCreateDirAndDeleteWithConnectionString() {
-
 	_require := require.New(s.T())
 	testName := s.T().Name()
 
@@ -202,6 +201,70 @@ func (s *RecordedTestSuite) TestGetAndCreateFileClient() {
 	_require.NoError(err)
 }
 
+func (s *RecordedTestSuite) TestCreateNewSubdirectoryClient() {
+	_require := require.New(s.T())
+	testName := s.T().Name()
+
+	accountName, _ := testcommon.GetGenericAccountInfo(testcommon.TestAccountDatalake)
+	_require.Greater(len(accountName), 0)
+
+	svcClient, err := testcommon.GetServiceClient(s.T(), testcommon.TestAccountDatalake, nil)
+	_require.NoError(err)
+
+	fsName := testcommon.GenerateFileSystemName(testName)
+	fsClient := svcClient.NewFileSystemClient(fsName)
+
+	_, err = fsClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	defer testcommon.DeleteFileSystem(context.Background(), _require, fsClient)
+
+	dirName := testcommon.GenerateDirName(testName)
+	dirClient := fsClient.NewDirectoryClient(dirName)
+
+	_, err = dirClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	subdirName := testcommon.GenerateSubDirName(testName)
+	subdirClient, err := dirClient.NewSubdirectoryClient(subdirName)
+	_require.NoError(err)
+
+	perm := "r-xr-x---"
+
+	_, err = subdirClient.Create(context.Background(), &directory.CreateOptions{
+		Permissions: &perm,
+		CPKInfo:     &testcommon.TestCPKByValue,
+	})
+	_require.NoError(err)
+
+	resp, err := subdirClient.GetProperties(context.Background(), &directory.GetPropertiesOptions{CPKInfo: &testcommon.TestCPKByValue})
+	_require.NoError(err)
+	_require.NotNil(resp.Permissions)
+	_require.Equal(perm, *resp.Permissions)
+	_require.Equal(*(resp.IsServerEncrypted), true)
+	_require.Equal(resp.EncryptionKeySHA256, testcommon.TestCPKByValue.EncryptionKeySHA256)
+
+	// Create a file under the new directory just to make sure we're not secretly targeting the parent
+	fileName := testcommon.GenerateFileName("newFile")
+	subdirFileClient, err := subdirClient.NewFileClient(fileName)
+	_require.NoError(err)
+
+	_, err = subdirFileClient.Create(context.Background(), nil)
+	_require.NoError(err)
+
+	fileContents := []byte(testcommon.DefaultData)
+	err = subdirFileClient.UploadBuffer(context.Background(), fileContents, nil)
+	_require.NoError(err)
+
+	// check for the file at the parent directory
+	dirFileClient, err := dirClient.NewFileClient(fileName)
+	_require.NoError(err)
+
+	_, err = dirFileClient.GetProperties(context.Background(), nil)
+	_require.Error(err) // we should get back a 404
+	_require.True(datalakeerror.HasCode(err, datalakeerror.PathNotFound))
+}
+
 func (s *RecordedTestSuite) TestCreateDirWithNilAccessConditions() {
 	_require := require.New(s.T())
 	testName := s.T().Name()
diff --git a/sdk/storage/azdatalake/internal/testcommon/common.go b/sdk/storage/azdatalake/internal/testcommon/common.go
index c8738d31fa43..4ea2bc4c1e6f 100644
--- a/sdk/storage/azdatalake/internal/testcommon/common.go
+++ b/sdk/storage/azdatalake/internal/testcommon/common.go
@@ -28,6 +28,7 @@ const (
 	FileSystemPrefix            = "gofs"
 	FilePrefix                  = "gotestfile"
 	DirPrefix                   = "gotestdir"
+	SubDirPrefix                = "gotestsubdir"
 	DefaultData                 = "Godatalakedata"
 	InvalidHeaderErrorSubstring = "invalid header field" // error thrown by the http client
 )
@@ -44,6 +45,10 @@ func GenerateDirName(testName string) string {
 	return DirPrefix + GenerateEntityName(testName)
 }
 
+func GenerateSubDirName(testName string) string {
+	return SubDirPrefix + GenerateEntityName(testName)
+}
+
 func GenerateEntityName(testName string) string {
 	return strings.ReplaceAll(strings.ReplaceAll(strings.ToLower(testName), "/", ""), "test", "")
 }